package eventBusV63

import (
	"errors"
	"sync"
	"time"
)

const (
	pubTimeout = time.Millisecond * 10
)

var (
	ErrPubsubTimeout = errors.New("failed to send message to topic because of timeout")
	ErrChannelFull   = errors.New("channel is full")
)

type Pubsub interface {
	Publish(topic string, msg interface{}) error
	Subscribe(topic string) (chan interface{}, error)
	Unsubscribe(topic string, ch chan interface{}) error
	Close() error
}

type pubsub struct {
	size     int
	channels map[string]map[chan interface{}]struct{}
	mu       sync.RWMutex
}

var channelPool *sync.Pool

// NewPubsub 初始化 pubsub 系统，使用默认的通道大小
func NewPubsub(size int) (Pubsub, error) {
	// 初始化通道池，使用给定的大小作为默认缓冲区大小
	channelPool = &sync.Pool{
		New: func() interface{} {
			return make(chan interface{}, size)
		},
	}

	return &pubsub{
		size:     size,
		channels: make(map[string]map[chan interface{}]struct{}),
	}, nil
}

// getChannelFromPool 从池中获取一个通道
func getChannelFromPool() chan interface{} {
	return channelPool.Get().(chan interface{})
}

// putChannelToPool 将通道放回池中，并清空其内容
func putChannelToPool(ch chan interface{}) {
	for len(ch) > 0 {
		<-ch
	}
	channelPool.Put(ch)
}

// Publish 向订阅者发送消息
func (m *pubsub) Publish(topic string, msg interface{}) error {
	m.mu.RLock() // 读锁，允许并发读取
	defer m.mu.RUnlock()

	if chs, ok := m.channels[topic]; ok {
		for ch := range chs {
			if err := m.publish(topic, ch, msg); err != nil {
				return err
			}
		}
	}
	return nil
}

// publish 尝试向单个通道发送消息，并处理通道扩容 
func (m *pubsub) publish(topic string, ch chan interface{}, msg interface{}) error { 
    // 尝试向现有通道发送消息 
    select { 
    case ch <- msg: // 尝试发送消息 
        return nil 
    default: 
        // 通道已满，需要动态扩容 
        newCh := make(chan interface{}, cap(ch)*2) // 扩容为原来的两倍 
 
        // 使用锁确保旧通道被正确替换为新通道 
        // m.mu.Lock()  
        // defer m.mu.Unlock()  
 
        // 将旧通道中的消息移动到新通道 
        go func() { 
            for { 
                select { 
                case v, ok := <-ch: 
                    if!ok { 
                        // 旧通道已关闭 
                        return 
                    } 
                    // 尝试将消息发送到新通道 
                    select { 
                    case newCh <- v: 
                        // 发送成功 
                    case <-time.After(pubTimeout): 
                        // 超时，放弃此消息 
                        return 
                    } 
                case <-time.After(pubTimeout): 
                    // 超时，退出循环 
                    return 
                } 
            } 
        }() 
 
        // 更新通道映射，指向新通道 
        if chs, ok := m.channels[topic];  ok { 
            delete(chs, ch)         // 删除旧通道 
            chs[newCh] = struct{}{} // 添加新扩容的通道 
        } 
 
        // 尝试向新扩容的通道发送消息 
        select { 
        case newCh <- msg: 
            return nil 
        case <-time.After(pubTimeout): 
            return ErrPubsubTimeout 
        } 
    } 
} 

// Subscribe 为给定主题创建一个新的订阅者通道
func (m *pubsub) Subscribe(topic string) (chan interface{}, error) {
	ch := getChannelFromPool() // 从池中获取一个通道

	m.mu.Lock()
	defer m.mu.Unlock()

	if _, ok := m.channels[topic]; !ok {
		m.channels[topic] = make(map[chan interface{}]struct{})
	}
	m.channels[topic][ch] = struct{}{} // 存储通道到主题的映射中
	return ch, nil
}

// Unsubscribe 从给定主题中移除订阅者通道
func (m *pubsub) Unsubscribe(topic string, ch chan interface{}) error {
	m.mu.Lock()
	defer m.mu.Unlock()

	if chs, ok := m.channels[topic]; ok {
		delete(chs, ch)      // 从主题的订阅者列表中移除通道
		putChannelToPool(ch) // 将通道放回池中
	}
	return nil
}

// Close 关闭 pubsub 系统，关闭所有通道并清理资源
func (m *pubsub) Close() error {
	m.mu.Lock()
	defer m.mu.Unlock()

	for topic, chs := range m.channels {
		for ch := range chs {
			close(ch)
		}
		delete(m.channels, topic)
	}
	return nil
}
